package com.ruoyi.mqtt.service;

import com.ruoyi.mqtt.model.SensorData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class DataProcessingService {

    private final ThreadPoolTaskExecutor threadPool;
    private final BlockingQueue<SensorData> dataQueue;
    private static final int BATCH_SIZE = 100;
    private static final int QUEUE_CAPACITY = 100000;
    private volatile boolean running = true;

    @Autowired
    private IoTDBService iotDBService;

    public DataProcessingService(@Qualifier("mqttThreadPool") ThreadPoolTaskExecutor mqttThreadPool) {
        this.threadPool = mqttThreadPool;
        this.dataQueue = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
    }

    @PostConstruct
    public void init() {
        // 初始化存储组
        iotDBService.initStorageGroup();

        // 启动批处理线程
        startBatchProcessor();
    }

    public void processData(SensorData data) {
        try {
            if (!dataQueue.offer(data, 100, TimeUnit.MILLISECONDS)) {
                log.warn("Queue is full, dropping data: {}", data);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Interrupted while offering data to queue", e);
        }
    }

    public void processBatch(List<SensorData> batch) {
        batch.forEach(this::processData);
    }

    private void startBatchProcessor() {
        Thread processorThread = new Thread(() -> {
            List<SensorData> batch = new ArrayList<>(BATCH_SIZE);

            while (running) {
                try {
                    SensorData data = dataQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (data != null) {
                        batch.add(data);

                        // 如果批次达到指定大小或队列为空且批次不为空，则处理批次
                        if (batch.size() >= BATCH_SIZE || (batch.size() > 0 && dataQueue.isEmpty())) {
                            final List<SensorData> currentBatch = new ArrayList<>(batch);
                            CompletableFuture.runAsync(() -> {
                                try {
                                    iotDBService.batchInsertSensorData(currentBatch);
                                } catch (Exception e) {
                                    log.error("Failed to process batch", e);
                                }
                            }, threadPool);

                            batch.clear();
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("Batch processor interrupted", e);
                    break;
                } catch (Exception e) {
                    log.error("Error in batch processor", e);
                }
            }

            // 处理剩余的数据
            if (!batch.isEmpty()) {
                try {
                    iotDBService.batchInsertSensorData(batch);
                } catch (Exception e) {
                    log.error("Failed to process final batch", e);
                }
            }
        }, "BatchProcessor");

        processorThread.setDaemon(true);
        processorThread.start();
    }

    @PreDestroy
    public void destroy() {
        log.info("Shutting down DataProcessingService...");
        running = false;

        // 等待剩余数据处理完成
        int waitCount = 0;
        while (!dataQueue.isEmpty() && waitCount < 30) {
            try {
                Thread.sleep(1000);
                waitCount++;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }

        if (!dataQueue.isEmpty()) {
            log.warn("{} messages still in queue after shutdown", dataQueue.size());
        }
    }
}
